package com.example.framework.mqtt.emqx;

import cn.hutool.json.JSONUtil;
import com.example.framework.mqtt.core.client.MqttPlatformClient;
import com.example.framework.mqtt.core.exception.MqttMessageException;
import com.example.framework.mqtt.core.model.MqttModel;
import net.dreamlu.iot.mqtt.codec.MqttProperties;
import net.dreamlu.iot.mqtt.codec.MqttQoS;
import net.dreamlu.iot.mqtt.core.client.IMqttClientMessageListener;
import net.dreamlu.iot.mqtt.core.client.MqttClient;
import net.dreamlu.iot.mqtt.core.client.MqttClientSubscription;
import org.apache.commons.lang3.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * <p>
 *
 * </p>
 *
 * @author sfjiang
 * @since 2024/11/26 17:44
 */
public class EmqxPlatformClient implements MqttPlatformClient {
    private final MqttClient mqttClient;

    public EmqxPlatformClient(MqttClient mqttClient) {
        this.mqttClient = mqttClient;
    }


    @Override
    public void publish(MqttModel<?> mqttModel) {
        validateTopic(mqttModel.getTopic());
        String dataJson = JSONUtil.toJsonStr(mqttModel.getData());
        this.mqttClient.publish(
                mqttModel.getTopic(),
                dataJson.getBytes(StandardCharsets.UTF_8),
                mqttModel.getQos(),
                mqttModel.isRetain()
        );
    }

    @Override
    public void publishWithResponseTopic(String responseTopic, MqttModel<?> mqttModel) throws MqttMessageException {
        validateTopic(mqttModel.getTopic());
        String dataJson = JSONUtil.toJsonStr(mqttModel.getData());
        MqttProperties properties = new MqttProperties();
        properties.add(new MqttProperties.StringProperty(MqttProperties.MqttPropertyType.RESPONSE_TOPIC.value(), responseTopic));
        this.mqttClient.publish(
                mqttModel.getTopic(),
                dataJson.getBytes(StandardCharsets.UTF_8),
                mqttModel.getQos(),
                mqttModel.isRetain(),
                properties
        );
    }

    @Override
    public void subscribeQos0(String topic, IMqttClientMessageListener messageListener) {
        this.mqttClient.subQos0(topic, messageListener);
    }

    @Override
    public void subscribeQos1(String topic, IMqttClientMessageListener messageListener) throws MqttMessageException {
        this.mqttClient.subQos1(topic, messageListener);
    }

    @Override
    public void subscribeQos2(String topic, IMqttClientMessageListener messageListener) throws MqttMessageException {
        this.mqttClient.subQos2(topic, messageListener);
    }

    @Override
    public void subscribe(String topic, MqttQoS qos, IMqttClientMessageListener messageListener) {
        this.mqttClient.subscribe(topic, qos, messageListener);
    }

    @Override
    public void subscribe(List<MqttClientSubscription> mqttListenerList) {
        this.mqttClient.subscribe(mqttListenerList);
    }

    @Override
    public void unsubscribe(String topic) {
        this.mqttClient.unSubscribe(topic);
    }

    @Override
    public void disconnect() {
        this.mqttClient.disconnect();
    }

    @Override
    public void reconnect() {
        this.mqttClient.reconnect();
    }

    private void validateTopic(String topic) throws MqttMessageException {
        if (StringUtils.isBlank(topic)) {
            throw new MqttMessageException("[EmqxPlatformClient] topic could not be blank!");
        }

    }
}
